{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "84b22f68",
   "metadata": {},
   "source": [
    "# @consumes basics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7a35b05c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | hide\n",
    "\n",
    "import asyncio\n",
    "import platform\n",
    "\n",
    "import asyncer\n",
    "from IPython.display import Markdown as md\n",
    "\n",
    "from fastkafka._components._subprocess import terminate_asyncio_process\n",
    "from fastkafka._testing.apache_kafka_broker import run_and_match\n",
    "from fastkafka.testing import ApacheKafkaBroker, run_script_and_cancel"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8c655e4f",
   "metadata": {},
   "source": [
    "You can use `@consumes` decorator to consume messages from Kafka topics. \n",
    "\n",
    "In this guide we will create a simple FastKafka app that will consume `HelloWorld` messages from hello_world topic."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18535f2f",
   "metadata": {},
   "source": [
    "## Import `FastKafka`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4ef969d",
   "metadata": {},
   "source": [
    "To use the `@consumes` decorator, first we need to import the base FastKafka app to create our application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d6a8fae",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from fastkafka import FastKafka\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "import_fastkafka = \"\"\"from fastkafka import FastKafka\n",
    "\"\"\"\n",
    "\n",
    "md(f\"```python\\n{import_fastkafka}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c2e9cb9f",
   "metadata": {},
   "source": [
    "In this demo we will log the messages to the output so that we can inspect and verify that our app is consuming properly. For that we need to import the logger."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a27e68b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from fastkafka._components.logger import get_logger\n",
       "\n",
       "logger = get_logger(__name__)\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "import_logger = \"\"\"from fastkafka._components.logger import get_logger\n",
    "\n",
    "logger = get_logger(__name__)\n",
    "\"\"\"\n",
    "\n",
    "md(f\"```python\\n{import_logger}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "09cb37e9",
   "metadata": {},
   "source": [
    "## Define the structure of the messages\n",
    "Next, you need to define the structure of the messages you want to consume from the topic using [pydantic](https://docs.pydantic.dev/). For the guide we'll stick to something basic, but you are free to define any complex message structure you wish in your project, just make sure it can be JSON encoded.\n",
    "\n",
    "Let's import `BaseModel` and `Field` from pydantic and create a simple `HelloWorld` class containing one string parameter `msg`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f83265a1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from pydantic import BaseModel, Field\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "import_pydantic = \"\"\"from pydantic import BaseModel, Field\n",
    "\"\"\"\n",
    "md(f\"```python\\n{import_pydantic}\\n```\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9c1e6ec8",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "class HelloWorld(BaseModel):\n",
       "    msg: str = Field(\n",
       "        ...,\n",
       "        example=\"Hello\",\n",
       "        description=\"Demo hello world message\",\n",
       "    )\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "define_HelloWorld = \"\"\"class HelloWorld(BaseModel):\n",
    "    msg: str = Field(\n",
    "        ...,\n",
    "        example=\"Hello\",\n",
    "        description=\"Demo hello world message\",\n",
    "    )\n",
    "\"\"\"\n",
    "md(f\"```python\\n{define_HelloWorld}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc1d810c",
   "metadata": {},
   "source": [
    "## Create a base FastKafka app\n",
    "\n",
    "Now we will create and define a base FastKafka app, replace the `<url_of_your_kafka_bootstrap_server>` and `<port_of_your_kafka_bootstrap_server>` with the actual values of your Kafka bootstrap server"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2732642f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "kafka_brokers = {\n",
       "    \"demo_broker\": {\n",
       "        \"url\": \"<url_of_your_kafka_bootstrap_server>\",\n",
       "        \"description\": \"local demo kafka broker\",\n",
       "        \"port\": \"<port_of_your_kafka_bootstrap_server>\",\n",
       "    }\n",
       "}\n",
       "\n",
       "app = FastKafka(kafka_brokers=kafka_brokers)\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "create_app = \"\"\"kafka_brokers = {\n",
    "    \"demo_broker\": {\n",
    "        \"url\": \"<url_of_your_kafka_bootstrap_server>\",\n",
    "        \"description\": \"local demo kafka broker\",\n",
    "        \"port\": \"<port_of_your_kafka_bootstrap_server>\",\n",
    "    }\n",
    "}\n",
    "\n",
    "app = FastKafka(kafka_brokers=kafka_brokers)\n",
    "\"\"\"\n",
    "md(f\"```python\\n{create_app}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6e41ebe6",
   "metadata": {},
   "source": [
    "## Create a consumer function and decorate it with `@consumes`\n",
    "\n",
    "Let's create a consumer function that will consume `HelloWorld` messages from *hello_world* topic and log them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "93dd7a11",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "@app.consumes()\n",
       "async def on_hello_world(msg: HelloWorld):\n",
       "    logger.info(f\"Got msg: {msg}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "decorate_consumes = \"\"\"@app.consumes()\n",
    "async def on_hello_world(msg: HelloWorld):\n",
    "    logger.info(f\"Got msg: {msg}\")\n",
    "\"\"\"\n",
    "md(f\"```python\\n{decorate_consumes}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9bdf5198",
   "metadata": {},
   "source": [
    "The function decorated with the `@consumes` decorator will be called when a message is produced to Kafka.\n",
    "\n",
    "The message will then be injected into the typed *msg* argument of the function and its type will be used to parse the message.\n",
    "\n",
    "In this example case, when the message is sent into a *hello_world* topic, it will be parsed into a HelloWorld class and `on_hello_world` function will be called with the parsed class as *msg* argument value."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b68c95f7",
   "metadata": {},
   "source": [
    "## Final app\n",
    "\n",
    "Your app code should look like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd7d88d3",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from fastkafka import FastKafka\n",
       "from pydantic import BaseModel, Field\n",
       "from fastkafka._components.logger import get_logger\n",
       "\n",
       "logger = get_logger(__name__)\n",
       "\n",
       "class HelloWorld(BaseModel):\n",
       "    msg: str = Field(\n",
       "        ...,\n",
       "        example=\"Hello\",\n",
       "        description=\"Demo hello world message\",\n",
       "    )\n",
       "\n",
       "kafka_brokers = {\n",
       "    \"demo_broker\": {\n",
       "        \"url\": \"<url_of_your_kafka_bootstrap_server>\",\n",
       "        \"description\": \"local demo kafka broker\",\n",
       "        \"port\": \"<port_of_your_kafka_bootstrap_server>\",\n",
       "    }\n",
       "}\n",
       "\n",
       "app = FastKafka(kafka_brokers=kafka_brokers)\n",
       "@app.consumes()\n",
       "async def on_hello_world(msg: HelloWorld):\n",
       "    logger.info(f\"Got msg: {msg}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_logger\n",
    "    + \"\\n\"\n",
    "    + define_HelloWorld\n",
    "    + \"\\n\"\n",
    "    + create_app\n",
    "    + decorate_consumes\n",
    ")\n",
    "md(f\"```python\\n{consumes_example}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8fa74215",
   "metadata": {},
   "source": [
    "## Run the app"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "afe529ab",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "Now we can run the app. Copy the code above in consumer_example.py and run it by running\n",
       "```shell\n",
       "fastkafka run --num-workers=1 --kafka-broker=demo_broker consumer_example:app\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "script_file = \"consumer_example.py\"\n",
    "filename = script_file.split(\".py\")[0]\n",
    "cmd = f\"fastkafka run --num-workers=1 --kafka-broker=demo_broker {filename}:app\"\n",
    "md(\n",
    "    f\"Now we can run the app. Copy the code above in {script_file} and run it by running\\n```shell\\n{cmd}\\n```\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e66be2b9",
   "metadata": {},
   "source": [
    "After running the command, you should see this output in your terminal:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "335ec62d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-06-15 07:15:54.167 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "23-06-15 07:15:54.168 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "23-06-15 07:15:54.169 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "23-06-15 07:15:54.170 [INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "23-06-15 07:15:54.170 [INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n",
      "23-06-15 07:15:54.171 [INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "23-06-15 07:15:54.171 [INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n",
      "23-06-15 07:15:54.172 [INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "23-06-15 07:15:55.001 [INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "23-06-15 07:15:57.462 [INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "23-06-15 07:15:59.432 [INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "23-06-15 07:15:59.432 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n",
      "23-06-15 07:16:05.941 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "23-06-15 07:16:05.942 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 13679...\n",
      "23-06-15 07:16:07.588 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 13679 terminated.\n",
      "23-06-15 07:16:07.588 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 13319...\n",
      "23-06-15 07:16:08.935 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 13319 terminated.\n",
      "23-06-15 07:16:08.940 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[\"hello_world\"], apply_nest_asyncio=True, listener_port=11092\n",
    ") as bootstrap_server:\n",
    "    server_url = bootstrap_server.split(\":\")[0]\n",
    "    server_port = bootstrap_server.split(\":\")[1]\n",
    "    exit_code, output = await run_script_and_cancel(\n",
    "        script=consumes_example.replace(\n",
    "            \"<url_of_your_kafka_bootstrap_server>\", server_url\n",
    "        ).replace(\"<port_of_your_kafka_bootstrap_server>\", server_port),\n",
    "        script_file=script_file,\n",
    "        cmd=cmd,\n",
    "        cancel_after=5,\n",
    "    )\n",
    "\n",
    "    expected_returncode = [0, 1]\n",
    "    assert exit_code in expected_returncode, exit_code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e35f6e21",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[14442]: 23-06-15 07:16:00.564 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[14442]: 23-06-15 07:16:00.564 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:9092'}\n",
      "[14442]: 23-06-15 07:16:00.577 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[14442]: 23-06-15 07:16:00.577 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'hello_world'})\n",
      "[14442]: 23-06-15 07:16:00.577 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'hello_world'}\n",
      "[14442]: 23-06-15 07:16:00.577 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[14442]: 23-06-15 07:16:00.585 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'hello_world': 1}. \n",
      "Starting process cleanup, this may take a few seconds...\n",
      "23-06-15 07:16:04.626 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 14442...\n",
      "[14442]: 23-06-15 07:16:05.735 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[14442]: 23-06-15 07:16:05.735 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-06-15 07:16:05.853 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 14442 terminated.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "print(output.decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d4ec6dab",
   "metadata": {},
   "source": [
    "## Send the message to kafka topic"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c81e65bc",
   "metadata": {},
   "source": [
    "Lets send a `HelloWorld` message to the *hello_world* topic and check if our consumer kafka application has logged the received message. In your terminal, run:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6ef181f6",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```shell\n",
       "echo { \\\"msg\\\": \\\"Hello world\\\" } | kafka-console-producer.sh --topic=hello_world --bootstrap-server=<addr_of_your_kafka_bootstrap_server>\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "script_extension = \".bat\" if platform.system() == \"Windows\" else \".sh\"\n",
    "escape_char = \"^\" if platform.system() == \"Windows\" else \"\\\\\"\n",
    "\n",
    "kafka_msg = 'echo {{ {escape_char}\"msg{escape_char}\": {escape_char}\"Hello world{escape_char}\" }}'.format(escape_char=escape_char)\n",
    "\n",
    "producer_cmd = f'{kafka_msg} | kafka-console-producer'+script_extension+' --topic=hello_world --bootstrap-server=<addr_of_your_kafka_bootstrap_server>'\n",
    "md(f\"```shell\\n{producer_cmd}\\n```\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a66904c8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-06-15 07:16:08.976 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "23-06-15 07:16:08.977 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "23-06-15 07:16:08.978 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "23-06-15 07:16:08.979 [INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "23-06-15 07:16:08.980 [INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "23-06-15 07:16:08.981 [INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "23-06-15 07:16:09.846 [INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "23-06-15 07:16:12.214 [INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "23-06-15 07:16:14.113 [INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "23-06-15 07:16:14.115 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n",
      "23-06-15 07:16:35.563 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "23-06-15 07:16:35.563 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 14824...\n",
      "23-06-15 07:16:37.212 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 14824 terminated.\n",
      "23-06-15 07:16:37.212 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 14462...\n",
      "23-06-15 07:16:38.560 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 14462 terminated.\n",
      "23-06-15 07:16:38.567 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_logger\n",
    "    + \"\\n\"\n",
    "    + define_HelloWorld\n",
    "    + \"\\n\"\n",
    "    + create_app\n",
    "    + decorate_consumes\n",
    ")\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[\"hello_world\"], apply_nest_asyncio=True, listener_port=11092\n",
    ") as bootstrap_server:\n",
    "    async with asyncer.create_task_group() as task_group:\n",
    "        server_url = bootstrap_server.split(\":\")[0]\n",
    "        server_port = bootstrap_server.split(\":\")[1]\n",
    "        consumer_task = task_group.soonify(run_script_and_cancel)(\n",
    "            script=consumes_example.replace(\n",
    "                \"<url_of_your_kafka_bootstrap_server>\", server_url\n",
    "            ).replace(\"<port_of_your_kafka_bootstrap_server>\", server_port),\n",
    "            script_file=script_file,\n",
    "            cmd=cmd,\n",
    "            cancel_after=20,\n",
    "        )\n",
    "        await asyncio.sleep(10)\n",
    "\n",
    "        producer_task = task_group.soonify(asyncio.create_subprocess_shell)(\n",
    "            cmd=producer_cmd.replace(\n",
    "                \"<addr_of_your_kafka_bootstrap_server>\", bootstrap_server\n",
    "            ),\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "        )\n",
    "\n",
    "assert \"Got msg: msg='Hello world'\" in consumer_task.value[1].decode(\"UTF-8\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e86e202e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[15588]: 23-06-15 07:16:15.282 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[15588]: 23-06-15 07:16:15.282 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:9092'}\n",
      "[15588]: 23-06-15 07:16:15.294 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[15588]: 23-06-15 07:16:15.294 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'hello_world'})\n",
      "[15588]: 23-06-15 07:16:15.295 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'hello_world'}\n",
      "[15588]: 23-06-15 07:16:15.295 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[15588]: 23-06-15 07:16:15.302 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'hello_world': 1}. \n",
      "[15588]: 23-06-15 07:16:25.867 [INFO] consumer_example: Got msg: msg='Hello world'\n",
      "Starting process cleanup, this may take a few seconds...\n",
      "23-06-15 07:16:34.168 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 15588...\n",
      "[15588]: 23-06-15 07:16:35.358 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[15588]: 23-06-15 07:16:35.359 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-06-15 07:16:35.475 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 15588 terminated.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "print(consumer_task.value[1].decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9292901c",
   "metadata": {},
   "source": [
    "You should see the \"Got msg: msg=\\'Hello world\\'\" being logged by your consumer."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "49b9c46e",
   "metadata": {},
   "source": [
    "## Choosing a topic"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed9e5f0c",
   "metadata": {},
   "source": [
    "You probably noticed that you didn't define which topic you are receiving the message from, this is because the `@consumes` decorator determines the topic by default from your function name.\n",
    "The decorator will take your function name and strip the default \"on_\" prefix from it and use the rest as the topic name. In this example case, the topic is *hello_world*."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dc3e9f8a",
   "metadata": {},
   "source": [
    "You can choose your custom prefix by defining the `prefix` parameter in consumes decorator, like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "da734ca4",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "@app.consumes(prefix=\"read_from_\")\n",
       "async def read_from_hello_world(msg: HelloWorld):\n",
       "    logger.info(f\"Got msg: {msg}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: False\n",
    "\n",
    "decorate_consumes_prefix = \"\"\"@app.consumes(prefix=\"read_from_\")\n",
    "async def read_from_hello_world(msg: HelloWorld):\n",
    "    logger.info(f\"Got msg: {msg}\")\n",
    "\"\"\"\n",
    "md(f\"```python\\n{decorate_consumes_prefix}\\n```\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b7563b45",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-06-15 07:17:44.438 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "23-06-15 07:17:44.439 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "23-06-15 07:17:44.440 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "23-06-15 07:17:44.440 [INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "23-06-15 07:17:44.441 [INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "23-06-15 07:17:44.442 [INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "23-06-15 07:17:45.272 [INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "23-06-15 07:17:47.608 [INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "23-06-15 07:17:49.517 [INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "23-06-15 07:17:49.518 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n",
      "23-06-15 07:18:10.834 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "23-06-15 07:18:10.834 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 16312...\n",
      "23-06-15 07:18:12.489 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 16312 terminated.\n",
      "23-06-15 07:18:12.490 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 15951...\n",
      "23-06-15 07:18:13.836 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 15951 terminated.\n",
      "23-06-15 07:18:13.845 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_logger\n",
    "    + define_HelloWorld\n",
    "    + create_app\n",
    "    + decorate_consumes_prefix\n",
    ")\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[\"hello_world\"], apply_nest_asyncio=True, listener_port=11092\n",
    ") as bootstrap_server:\n",
    "    async with asyncer.create_task_group() as task_group:\n",
    "        server_url = bootstrap_server.split(\":\")[0]\n",
    "        server_port = bootstrap_server.split(\":\")[1]\n",
    "        consumer_task = task_group.soonify(run_script_and_cancel)(\n",
    "            script=consumes_example.replace(\n",
    "                \"<url_of_your_kafka_bootstrap_server>\", server_url\n",
    "            ).replace(\"<port_of_your_kafka_bootstrap_server>\", server_port),\n",
    "            script_file=script_file,\n",
    "            cmd=cmd,\n",
    "            cancel_after=20,\n",
    "        )\n",
    "        await asyncio.sleep(10)\n",
    "\n",
    "        producer_task = task_group.soonify(asyncio.create_subprocess_shell)(\n",
    "            cmd=producer_cmd.replace(\n",
    "                \"<addr_of_your_kafka_bootstrap_server>\", bootstrap_server\n",
    "            ),\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "        )\n",
    "\n",
    "assert \"Got msg: msg='Hello world'\" in consumer_task.value[1].decode(\"UTF-8\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0c4225dd",
   "metadata": {},
   "source": [
    "Also, you can define the topic name completely by defining the `topic` in parameter in consumes decorator, like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9bf881b6",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "@app.consumes(topic=\"my_special_topic\")\n",
       "async def on_hello_world(msg: HelloWorld):\n",
       "    logger.info(f\"Got msg: {msg}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "decorate_consumes_topic = \"\"\"@app.consumes(topic=\"my_special_topic\")\n",
    "async def on_hello_world(msg: HelloWorld):\n",
    "    logger.info(f\"Got msg: {msg}\")\n",
    "\"\"\"\n",
    "md(f\"```python\\n{decorate_consumes_topic}\\n```\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "612baa8c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-06-15 07:18:13.874 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "23-06-15 07:18:13.874 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "23-06-15 07:18:13.875 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "23-06-15 07:18:13.877 [INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "23-06-15 07:18:13.878 [INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "23-06-15 07:18:13.879 [INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "23-06-15 07:18:14.717 [INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "23-06-15 07:18:17.045 [INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "23-06-15 07:18:18.971 [INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "23-06-15 07:18:18.972 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n",
      "23-06-15 07:18:40.425 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "23-06-15 07:18:40.426 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 17799...\n",
      "23-06-15 07:18:42.046 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 17799 terminated.\n",
      "23-06-15 07:18:42.046 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 17438...\n",
      "23-06-15 07:18:43.396 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 17438 terminated.\n",
      "23-06-15 07:18:43.401 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_logger\n",
    "    + define_HelloWorld\n",
    "    + create_app\n",
    "    + decorate_consumes_topic\n",
    ")\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[\"my_special_topic\"], apply_nest_asyncio=True, listener_port=11092\n",
    ") as bootstrap_server:\n",
    "    async with asyncer.create_task_group() as task_group:\n",
    "        server_url = bootstrap_server.split(\":\")[0]\n",
    "        server_port = bootstrap_server.split(\":\")[1]\n",
    "        consumer_task = task_group.soonify(run_script_and_cancel)(\n",
    "            script=consumes_example.replace(\n",
    "                \"<url_of_your_kafka_bootstrap_server>\", server_url\n",
    "            ).replace(\"<port_of_your_kafka_bootstrap_server>\", server_port),\n",
    "            script_file=script_file,\n",
    "            cmd=cmd,\n",
    "            cancel_after=20,\n",
    "        )\n",
    "        await asyncio.sleep(10)\n",
    "\n",
    "        producer_task = task_group.soonify(asyncio.create_subprocess_shell)(\n",
    "            cmd=producer_cmd.replace(\n",
    "                \"<addr_of_your_kafka_bootstrap_server>\", bootstrap_server\n",
    "            ).replace(\"topic=hello_world\", \"topic=my_special_topic\"),\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "        )\n",
    "\n",
    "assert \"Got msg: msg='Hello world'\" in consumer_task.value[1].decode(\n",
    "    \"UTF-8\"\n",
    "), consumer_task.value[1].decode(\"UTF-8\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "da50972c",
   "metadata": {},
   "source": [
    "## Message data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7e4e8052",
   "metadata": {},
   "source": [
    "The message received from kafka is translated from binary JSON representation int the class defined by typing of *msg* parameter in the function decorated by the `@consumes` decorator.\n",
    "\n",
    "In this example case, the message will be parsed into a `HelloWorld` class."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8763a137",
   "metadata": {},
   "source": [
    "## Message metadata\n",
    "\n",
    "If you need any of Kafka message metadata such as timestamp, partition or headers you can access the metadata by adding a EventMetadata typed argument to your consumes function and the metadata from the incoming message will be automatically injected when calling the consumes function.\n",
    "\n",
    "Let's demonstrate that."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cbde9875",
   "metadata": {},
   "source": [
    "### Create a consumer function with metadata\n",
    "\n",
    "The only difference from the original basic consume function is that we are now passing the `meta: EventMetadata` argument to the function. The `@consumes` decorator will register that and, when a message is consumed, it will also pass the metadata to your function. Now you can use the metadata in your consume function. \n",
    "Lets log it to see what it contains."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "206a4c79",
   "metadata": {},
   "source": [
    "First, we need to import the EventMetadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9bc607d9",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from fastkafka import EventMetadata\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "import_event_metadata = \"\"\"from fastkafka import EventMetadata\n",
    "\"\"\"\n",
    "md(f\"```python\\n{import_event_metadata}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "006804ea",
   "metadata": {},
   "source": [
    "Now we can add the `meta` argument to our consuming function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e3d029e0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "@app.consumes()\n",
       "async def on_hello_world(msg: HelloWorld, meta: EventMetadata):\n",
       "    logger.info(f\"Got metadata: {meta}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "decorate_consumes_meta = \"\"\"@app.consumes()\n",
    "async def on_hello_world(msg: HelloWorld, meta: EventMetadata):\n",
    "    logger.info(f\"Got metadata: {meta}\")\n",
    "\"\"\"\n",
    "md(f\"```python\\n{decorate_consumes_meta}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d9df2e13",
   "metadata": {},
   "source": [
    "Your final app should look like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "90b842f7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from fastkafka import FastKafka\n",
       "from pydantic import BaseModel, Field\n",
       "from fastkafka import EventMetadata\n",
       "from fastkafka._components.logger import get_logger\n",
       "\n",
       "logger = get_logger(__name__)\n",
       "class HelloWorld(BaseModel):\n",
       "    msg: str = Field(\n",
       "        ...,\n",
       "        example=\"Hello\",\n",
       "        description=\"Demo hello world message\",\n",
       "    )\n",
       "kafka_brokers = {\n",
       "    \"demo_broker\": {\n",
       "        \"url\": \"<url_of_your_kafka_bootstrap_server>\",\n",
       "        \"description\": \"local demo kafka broker\",\n",
       "        \"port\": \"<port_of_your_kafka_bootstrap_server>\",\n",
       "    }\n",
       "}\n",
       "\n",
       "app = FastKafka(kafka_brokers=kafka_brokers)\n",
       "@app.consumes()\n",
       "async def on_hello_world(msg: HelloWorld, meta: EventMetadata):\n",
       "    logger.info(f\"Got metadata: {meta}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_event_metadata\n",
    "    + import_logger\n",
    "    + define_HelloWorld\n",
    "    + create_app\n",
    "    + decorate_consumes_meta\n",
    ")\n",
    "\n",
    "md(f\"```python\\n{consumes_example}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9143036a",
   "metadata": {},
   "source": [
    "Now lets run the app and send a message to the broker to see the logged message metadata."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a6d637ad",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-06-15 07:18:49.398 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "23-06-15 07:18:49.399 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "23-06-15 07:18:49.400 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "23-06-15 07:18:49.401 [INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "23-06-15 07:18:49.401 [INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "23-06-15 07:18:49.402 [INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "23-06-15 07:18:50.243 [INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "23-06-15 07:18:52.575 [INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "23-06-15 07:18:54.527 [INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "23-06-15 07:18:54.528 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n",
      "23-06-15 07:19:15.835 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "23-06-15 07:19:15.836 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 19287...\n",
      "23-06-15 07:19:17.454 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 19287 terminated.\n",
      "23-06-15 07:19:17.455 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 18925...\n",
      "23-06-15 07:19:18.802 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 18925 terminated.\n",
      "23-06-15 07:19:18.810 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[\"hello_world\"], apply_nest_asyncio=True, listener_port=11092\n",
    ") as bootstrap_server:\n",
    "    async with asyncer.create_task_group() as task_group:\n",
    "        server_url = bootstrap_server.split(\":\")[0]\n",
    "        server_port = bootstrap_server.split(\":\")[1]\n",
    "        consumer_task = task_group.soonify(run_script_and_cancel)(\n",
    "            script=consumes_example.replace(\n",
    "                \"<url_of_your_kafka_bootstrap_server>\", server_url\n",
    "            ).replace(\"<port_of_your_kafka_bootstrap_server>\", server_port),\n",
    "            script_file=script_file,\n",
    "            cmd=cmd,\n",
    "            cancel_after=20,\n",
    "        )\n",
    "        await asyncio.sleep(10)\n",
    "\n",
    "        producer_task = task_group.soonify(asyncio.create_subprocess_shell)(\n",
    "            cmd=producer_cmd.replace(\n",
    "                \"<addr_of_your_kafka_bootstrap_server>\", bootstrap_server\n",
    "            ),\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "        )\n",
    "\n",
    "assert \"Got metadata: \" in consumer_task.value[1].decode(\n",
    "    \"UTF-8\"\n",
    "), consumer_task.value[1].decode(\"UTF-8\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "74082867",
   "metadata": {},
   "source": [
    "You should see a similar log as the one below and the metadata being logged in your app."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f4e6d4c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[20050]: 23-06-15 07:18:55.661 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[20050]: 23-06-15 07:18:55.661 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:9092'}\n",
      "[20050]: 23-06-15 07:18:55.675 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[20050]: 23-06-15 07:18:55.675 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'hello_world'})\n",
      "[20050]: 23-06-15 07:18:55.675 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'hello_world'}\n",
      "[20050]: 23-06-15 07:18:55.675 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[20050]: 23-06-15 07:18:55.682 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'hello_world': 1}. \n",
      "[20050]: 23-06-15 07:19:06.337 [INFO] consumer_example: Got metadata: EventMetadata(topic='hello_world', partition=0, offset=0, timestamp=1686813546255, timestamp_type=0, key=None, value=b'{ \"msg\": \"Hello world\" }', checksum=None, serialized_key_size=-1, serialized_value_size=24, headers=())\n",
      "Starting process cleanup, this may take a few seconds...\n",
      "23-06-15 07:19:14.547 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 20050...\n",
      "[20050]: 23-06-15 07:19:15.630 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[20050]: 23-06-15 07:19:15.630 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-06-15 07:19:15.742 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 20050 terminated.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "print(consumer_task.value[1].decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5252b37a",
   "metadata": {},
   "source": [
    "As you can see in the log, from the metadata you now have the information about the partition, offset, timestamp, key and headers. :tada:"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46bb2a38",
   "metadata": {},
   "source": [
    "## Dealing with high latency consuming functions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2441facc",
   "metadata": {},
   "source": [
    "If your functions have high latency due to, for example, lengthy database calls you will notice a big decrease in performance. This is due to the issue of how the consumes decorator executes your consume functions when consuming events. By default, the consume function will run the consuming funtions for one topic sequentially, this is the most straightforward approach and results with the least amount of overhead.\n",
    "\n",
    "But, to handle those high latency tasks and run them in parallel, FastKafka has a `DynamicTaskExecutor` prepared for your consumers. This executor comes with additional overhead, so use it only when you need to handle high latency functions.\n",
    "\n",
    "Lets demonstrate how to use it.\n",
    "\n",
    "To your consumes decorator, add an `executor` option and set it to `\"DynamicTaskExecutor\"`, this will enable the consumer to handle high latency functions effectively.\n",
    "\n",
    "Your consuming function should now look like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "90e6743b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "@app.consumes(executor=\"DynamicTaskExecutor\")\n",
       "async def on_hello_world(msg: HelloWorld):\n",
       "    logger.info(f\"Got msg: {msg}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "decorate_consumes_executor = \"\"\"@app.consumes(executor=\"DynamicTaskExecutor\")\n",
    "async def on_hello_world(msg: HelloWorld):\n",
    "    logger.info(f\"Got msg: {msg}\")\n",
    "\"\"\"\n",
    "md(f\"```python\\n{decorate_consumes_executor}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d0435fcf",
   "metadata": {},
   "source": [
    "And the complete app should now look like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d01ad105",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "from fastkafka import FastKafka\n",
       "from pydantic import BaseModel, Field\n",
       "from fastkafka._components.logger import get_logger\n",
       "\n",
       "logger = get_logger(__name__)\n",
       "\n",
       "class HelloWorld(BaseModel):\n",
       "    msg: str = Field(\n",
       "        ...,\n",
       "        example=\"Hello\",\n",
       "        description=\"Demo hello world message\",\n",
       "    )\n",
       "\n",
       "kafka_brokers = {\n",
       "    \"demo_broker\": {\n",
       "        \"url\": \"<url_of_your_kafka_bootstrap_server>\",\n",
       "        \"description\": \"local demo kafka broker\",\n",
       "        \"port\": \"<port_of_your_kafka_bootstrap_server>\",\n",
       "    }\n",
       "}\n",
       "\n",
       "app = FastKafka(kafka_brokers=kafka_brokers)\n",
       "\n",
       "@app.consumes(executor=\"DynamicTaskExecutor\")\n",
       "async def on_hello_world(msg: HelloWorld):\n",
       "    logger.info(f\"Got msg: {msg}\")\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_logger\n",
    "    + \"\\n\"\n",
    "    + define_HelloWorld\n",
    "    + \"\\n\"\n",
    "    + create_app\n",
    "    + \"\\n\"\n",
    "    + decorate_consumes_executor\n",
    ")\n",
    "\n",
    "md(f\"```python\\n{consumes_example}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4293d59e",
   "metadata": {},
   "source": [
    "You can now run your app using the CLI commands described in this guide.\n",
    "\n",
    "Lets send a `HelloWorld` message to the *hello_world* topic and check if our consumer kafka application has logged the received message. In your terminal, run:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fedc1e5c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```shell\n",
       "echo { \\\"msg\\\": \\\"Hello world\\\" } | kafka-console-producer.sh --topic=hello_world --bootstrap-server=<addr_of_your_kafka_bootstrap_server>\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "script_extension = \".bat\" if platform.system() == \"Windows\" else \".sh\"\n",
    "escape_char = \"^\" if platform.system() == \"Windows\" else \"\\\\\"\n",
    "\n",
    "kafka_msg = 'echo {{ {escape_char}\"msg{escape_char}\": {escape_char}\"Hello world{escape_char}\" }}'.format(escape_char=escape_char)\n",
    "\n",
    "\n",
    "producer_cmd = f'{kafka_msg} | kafka-console-producer'+script_extension+' --topic=hello_world --bootstrap-server=<addr_of_your_kafka_bootstrap_server>'\n",
    "md(f\"```shell\\n{producer_cmd}\\n```\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "53c47569",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-06-15 07:19:18.845 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "23-06-15 07:19:18.847 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "23-06-15 07:19:18.847 [WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "23-06-15 07:19:18.848 [INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "23-06-15 07:19:18.849 [INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "23-06-15 07:19:18.849 [INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "23-06-15 07:19:19.677 [INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "23-06-15 07:19:22.048 [INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "23-06-15 07:19:23.954 [INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "23-06-15 07:19:23.955 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n",
      "23-06-15 07:19:45.396 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "23-06-15 07:19:45.397 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 20774...\n",
      "23-06-15 07:19:47.050 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 20774 terminated.\n",
      "23-06-15 07:19:47.050 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 20413...\n",
      "23-06-15 07:19:48.397 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 20413 terminated.\n",
      "23-06-15 07:19:48.404 [INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "\n",
    "consumes_example = (\n",
    "    import_fastkafka\n",
    "    + import_pydantic\n",
    "    + import_logger\n",
    "    + define_HelloWorld\n",
    "    + create_app\n",
    "    + decorate_consumes_executor\n",
    ")\n",
    "\n",
    "with ApacheKafkaBroker(\n",
    "    topics=[\"hello_world\"], apply_nest_asyncio=True, listener_port=11092\n",
    ") as bootstrap_server:\n",
    "    async with asyncer.create_task_group() as task_group:\n",
    "        server_url = bootstrap_server.split(\":\")[0]\n",
    "        server_port = bootstrap_server.split(\":\")[1]\n",
    "        consumer_task = task_group.soonify(run_script_and_cancel)(\n",
    "            script=consumes_example.replace(\n",
    "                \"<url_of_your_kafka_bootstrap_server>\", server_url\n",
    "            ).replace(\"<port_of_your_kafka_bootstrap_server>\", server_port),\n",
    "            script_file=script_file,\n",
    "            cmd=cmd,\n",
    "            cancel_after=20,\n",
    "        )\n",
    "        await asyncio.sleep(10)\n",
    "\n",
    "        producer_task = task_group.soonify(asyncio.create_subprocess_shell)(\n",
    "            cmd=producer_cmd.replace(\n",
    "                \"<addr_of_your_kafka_bootstrap_server>\", bootstrap_server\n",
    "            ),\n",
    "            stdout=asyncio.subprocess.PIPE,\n",
    "            stderr=asyncio.subprocess.PIPE,\n",
    "        )\n",
    "\n",
    "assert \"Got msg: msg='Hello world'\" in consumer_task.value[1].decode(\"UTF-8\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8647faa",
   "metadata": {},
   "source": [
    "You should see a similar log as the one below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aea11db6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[21539]: 23-06-15 07:19:25.135 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[21539]: 23-06-15 07:19:25.135 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:9092'}\n",
      "[21539]: 23-06-15 07:19:25.147 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[21539]: 23-06-15 07:19:25.147 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'hello_world'})\n",
      "[21539]: 23-06-15 07:19:25.147 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'hello_world'}\n",
      "[21539]: 23-06-15 07:19:25.147 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[21539]: 23-06-15 07:19:25.154 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'hello_world': 1}. \n",
      "[21539]: 23-06-15 07:19:35.512 [INFO] consumer_example: Got msg: msg='Hello world'\n",
      "Starting process cleanup, this may take a few seconds...\n",
      "23-06-15 07:19:44.023 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 21539...\n",
      "[21539]: 23-06-15 07:19:45.202 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[21539]: 23-06-15 07:19:45.203 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-06-15 07:19:45.313 [INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 21539 terminated.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "print(consumer_task.value[1].decode(\"UTF-8\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9572ebf3",
   "metadata": {},
   "source": [
    "Inside the log, you should see the \"Got msg: msg=\\'Hello world\\'\" being logged by your consumer."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
